package com.atguigu.flink.chapter06;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;
import java.util.Random;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/22 13:52
 */
public class Flink07_MarkettingAnalysisByChannel {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1.读取数据、转换成 POJO
        env
                .addSource(new MySource())
                .keyBy(new KeySelector<MarketingUserBehavior, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> getKey(MarketingUserBehavior value) throws Exception {
                        return Tuple2.of(value.getChannel(), value.getBehavior());
                    }
                })    // 按照统计维度分组：渠道、行为
                .process(new KeyedProcessFunction<Tuple2<String,String>, MarketingUserBehavior, String>() {
                    // 定义一个键控状态（分组隔离，每个分组一个状态实例）
                    @Override
                    public void processElement(MarketingUserBehavior value, Context ctx, Collector<String> out) throws Exception {
                        // 每来一条数据，状态值+1
                    }
                })
                .print();


        env.execute();
    }

    public static class MySource implements SourceFunction<MarketingUserBehavior> {

        private volatile boolean isRunning = true;
        private Random random = new Random();
        private List<String> behaviorList = Arrays.asList("DOWNLOAD", "INSTALL", "UPDATE", "UNINSTALL");
        private List<String> channelList = Arrays.asList("XIAOMI", "HUAWEI", "OPPO", "VIVO", "APPSTORE");

        @Override
        public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
            while (isRunning) {
                MarketingUserBehavior marketingUserBehavior = new MarketingUserBehavior(
                        random.nextLong(),
                        behaviorList.get(random.nextInt(behaviorList.size())),
                        channelList.get(random.nextInt(channelList.size())),
                        System.currentTimeMillis()
                );
                ctx.collect(marketingUserBehavior);
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
